package com.jzlw.cloud.platform.apple.httpsourceexpress;/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

import com.google.common.base.Preconditions;
import com.google.common.base.Throwables;
import org.apache.commons.lang.StringUtils;
import org.apache.flume.ChannelException;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.EventDrivenSource;
import org.apache.flume.conf.Configurable;
import org.apache.flume.instrumentation.SourceCounter;
import org.apache.flume.source.AbstractSource;
import org.apache.flume.source.http.HTTPBadRequestException;
import org.apache.flume.source.http.HTTPSource;
import org.apache.flume.source.http.HTTPSourceConfigurationConstants;
import org.apache.flume.source.http.HTTPSourceHandler;
import org.apache.flume.tools.HTTPServerConstraintUtil;
import org.mortbay.jetty.Connector;
import org.mortbay.jetty.Server;
import org.mortbay.jetty.nio.SelectChannelConnector;
import org.mortbay.jetty.security.SslSocketConnector;
import org.mortbay.jetty.servlet.ServletHolder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.net.ssl.SSLServerSocket;
import javax.servlet.ServletOutputStream;
import javax.servlet.http.HttpServlet;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.io.IOException;
import java.net.ServerSocket;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;

/**
 * A source which accepts Flume Events by HTTP POST and GET. GET should be used
 * for experimentation only. HTTP requests are converted into flume events by a
 * pluggable "handler" which must implement the
 * {@linkplain HTTPSourceHandler} interface. This handler takes a
 * {@linkplain HttpServletRequest} and returns a list of flume events.
 * <p>
 * The source accepts the following parameters: <p> <tt>port</tt>: port to which
 * the server should bind. Mandatory <p> <tt>handler</tt>: the class that
 * deserializes a HttpServletRequest into a list of flume events. This class
 * must implement HTTPSourceHandler. Default:
 * {@linkplain "JSONHandler"}. <p> <tt>handler.*</tt> Any configuration
 * to be passed to the handler. <p>
 * <p>
 * All events deserialized from one Http request are committed to the channel in
 * one transaction, thus allowing for increased efficiency on channels like the
 * file channel. If the handler throws an exception this source will return
 * a HTTP status of 400. If the channel is full, or the source is unable to
 * append events to the channel, the source will return a HTTP 503 - Temporarily
 * unavailable status.
 * <p>
 * A JSON handler which converts JSON objects to Flume evnts is provided.
 */
public class HttpSourceExpress extends AbstractSource implements
        EventDrivenSource, Configurable {
  /*
   * There are 2 ways of doing this:
   * a. Have a static server instance and use connectors in each source
   *    which binds to the port defined for that source.
   * b. Each source starts its own server instance, which binds to the source's
   *    port.
   *
   * b is more efficient than a because Jetty does not allow binding a
   * servlet to a connector. So each request will need to go through each
   * each of the handlers/servlet till the correct one is found.
   *
   */

    private static final Logger LOG = LoggerFactory.getLogger(HTTPSource.class);
    /**
     * 监听的端口
     */
    private volatile Integer port;
    /**
     * 监听器的名字
     */
    private volatile Server srv;
    /**
     * 监听的地址
     */
    private volatile String host;
    /**
     * http的解析器
     */
    private BlobHandlerExpress handler;
    /**
     * 返回的状态码
     */
    private int status = 0;

    /**
     * 返回的错误消息
     */
    private String errMsg;

    /**
     * 一共接收数据量
     */
    private int totalAccessedNum;

    /**
     * 验证成功数据量
     */
    private int successAccessedNum;

    /**
     * json验证错误信息集合
     */
    private String recordErrMsg;
    /**
     * 全局对象
     */
    private SourceCounter sourceCounter;

    // SSL configuration variable
    /**
     * keyStore文件保存的地址
     */
    private volatile String keyStorePath;
    /**
     * keyStore打开的密码
     */
    private volatile String keyStorePassword;
    /**
     * 是否使用ssl加密
     */
    private volatile Boolean sslEnabled;
    private final List<String> excludedProtocols = new LinkedList<String>();

    /**
     * 初始化配置
     */
    public void configure(Context context) {
        try {
            // SSL related config
            //是否使用ssl加密连接
            sslEnabled = context.getBoolean(HTTPSourceConfigurationConstants.SSL_ENABLED, false);
            //监听的端口
            port = context.getInteger(HTTPSourceConfigurationConstants.CONFIG_PORT);
            //监听的地址
            host = context.getString(HTTPSourceConfigurationConstants.CONFIG_BIND,
                    HTTPSourceConfigurationConstants.DEFAULT_BIND);
            //执行检查
            Preconditions.checkState(host != null && StringUtils.isNotBlank(host),
                    "HTTPSource hostname specified is empty");
            Preconditions.checkNotNull(port, "HTTPSource requires a port number to be"
                    + " specified");
            //获得http的解析类，如果没有使用默认的解析类
            String handlerClassName = context.getString(
                    HTTPSourceConfigurationConstants.CONFIG_HANDLER,
                    HTTPSourceConfigurationConstants.DEFAULT_HANDLER).trim();
            //是否使用ssl加密，如果使用读取keyStore的地址
            if (sslEnabled) {
                LOG.debug("SSL configuration enabled");
                keyStorePath = context.getString(HTTPSourceConfigurationConstants.SSL_KEYSTORE);
                Preconditions.checkArgument(keyStorePath != null && StringUtils.isNotBlank(keyStorePath),
                        "Keystore is required for SSL Conifguration");
                keyStorePassword =
                        context.getString(HTTPSourceConfigurationConstants.SSL_KEYSTORE_PASSWORD);
                Preconditions.checkArgument(keyStorePassword != null,
                        "Keystore password is required for SSL Configuration");
                String excludeProtocolsStr =
                        context.getString(HTTPSourceConfigurationConstants.EXCLUDE_PROTOCOLS);
                if (excludeProtocolsStr == null) {
                    excludedProtocols.add("SSLv3");
                } else {
                    excludedProtocols.addAll(Arrays.asList(excludeProtocolsStr.split(" ")));
                    if (!excludedProtocols.contains("SSLv3")) {
                        excludedProtocols.add("SSLv3");
                    }
                }
            }


            @SuppressWarnings("unchecked")
            /**
             *实例化http解析类
             */
                    Class<? extends HTTPSourceHandler> clazz =
                    (Class<? extends HTTPSourceHandler>)
                            Class.forName(handlerClassName);
            try {
                handler = (BlobHandlerExpress) clazz.getDeclaredConstructor().newInstance();
            } catch (Exception e) {
                //3表示实例化BlobHandlerMySelf对象发生错误
                status = 3;
                errMsg = "实例化BlobHandlerMySelf对象发生错误";
                LOG.error("实例化BlobHandlerMySelf对象发生错误", e);
                throw new Exception("实例化BlobHandlerMySelf对象发生错误", e);
            }
            //ref: http://docs.codehaus.org/display/JETTY/Embedding+Jetty
            //ref: http://jetty.codehaus.org/jetty/jetty-6/apidocs/org/mortbay/jetty/servlet/Context.html
            Map<String, String> subProps =
                    context.getSubProperties(
                            HTTPSourceConfigurationConstants.CONFIG_HANDLER_PREFIX);
            handler.configure(new Context(subProps));
        } catch (ClassNotFoundException ex) {
            if (StringUtils.isBlank(errMsg)) {
                errMsg = "Error while configuring HTTPSource. Exception follows.";
            }
            LOG.error("Error while configuring HTTPSource. Exception follows.", ex);
            Throwables.propagate(ex);
        } catch (ClassCastException ex) {
            if (StringUtils.isBlank(errMsg)) {
                errMsg = "Deserializer is not an instance of HTTPSourceHandler."
                        + "Deserializer must implement HTTPSourceHandler.";
            }
            LOG.error("Deserializer is not an instance of HTTPSourceHandler."
                    + "Deserializer must implement HTTPSourceHandler.");
            Throwables.propagate(ex);
        } catch (Exception ex) {
            if (StringUtils.isBlank(errMsg)) {
                errMsg = "Error configuring HTTPSource!";
            }
            LOG.error("Error configuring HTTPSource!", ex);
            Throwables.propagate(ex);
        }
        if (sourceCounter == null) {
            sourceCounter = new SourceCounter(getName());
        }
    }

    private void checkHostAndPort() {
        Preconditions.checkState(host != null && StringUtils.isNotBlank(host),
                "HTTPSource hostname specified is empty");
        Preconditions.checkNotNull(port, "HTTPSource requires a port number to be"
                + " specified");
    }

    /**
     * 开始监听
     */
    @Override
    public void start() {
        Preconditions.checkState(srv == null,
                "Running HTTP Server found in source: " + getName()
                        + " before I started one."
                        + "Will not attempt to start.");
        srv = new Server();

        // Connector Array
        Connector[] connectors = new Connector[1];

        if (sslEnabled) {
            SslSocketConnector sslSocketConnector = new HTTPSourceSocketConnector(excludedProtocols);
            sslSocketConnector.setKeystore(keyStorePath);
            sslSocketConnector.setKeyPassword(keyStorePassword);
            sslSocketConnector.setReuseAddress(true);
            connectors[0] = sslSocketConnector;
        } else {
            SelectChannelConnector connector = new SelectChannelConnector();
            connector.setReuseAddress(true);
            connectors[0] = connector;
        }

        connectors[0].setHost(host);
        connectors[0].setPort(port);
        srv.setConnectors(connectors);
        try {
            org.mortbay.jetty.servlet.Context root = new org.mortbay.jetty.servlet.Context(
                    srv, "/", org.mortbay.jetty.servlet.Context.SESSIONS);
            root.addServlet(new ServletHolder(new FlumeHTTPServlet()), "/");
            HTTPServerConstraintUtil.enforceConstraints(root);
            srv.start();
            Preconditions.checkArgument(srv.getHandler().equals(root));
        } catch (Exception ex) {
            //4表示在启动HttpSource的过程中发生异常
            status = 4;
            if (StringUtils.isBlank(errMsg)) {
                errMsg = "启动HttpSource的过程中发生异常";
            }
            LOG.error("Error while starting HTTPSource. Exception follows.", ex);
            Throwables.propagate(ex);
        }
        Preconditions.checkArgument(srv.isRunning());
        sourceCounter.start();
        super.start();
    }

    @Override
    public void stop() {
        try {
            srv.stop();
            srv.join();
            srv = null;
        } catch (Exception ex) {
            status = 5;
            if (StringUtils.isBlank(errMsg)) {
                errMsg = "关闭HttpSource的过程中发生异常！";
            }
            LOG.error("Error while stopping HTTPSource. Exception follows.", ex);
        }
        sourceCounter.stop();
        LOG.info("Http source {} stopped. Metrics: {}", getName(), sourceCounter);
    }

    private class FlumeHTTPServlet extends HttpServlet {

        private static final long serialVersionUID = 4891924863218790344L;

        @Override
        public void doPost(HttpServletRequest request, HttpServletResponse response)
                throws IOException {
            List<Event> events = Collections.emptyList(); //create empty list
            ServletOutputStream out = response.getOutputStream();
            try {
                Map<String, Object> inMap = handler.getMyselfMothed(request);
                if (status == 0) {
                    status = (Integer) inMap.get("status");
                }
                if (StringUtils.isBlank(errMsg)) {
                    errMsg = (String) inMap.get("errMsg");
                }

                //如果遇到致命异常，直接返回
                if (status == 8 || status == 7 || status == 2
                        || status == 1 || status == 11
                        || status == 13 || status == 14) {
                    LOG.error(errMsg);
                    finalMhd(request,response,out);
                    return;
                }
                //获取总接收条数，成功条数，失败条数
                totalAccessedNum=(Integer) inMap.get("totalAccessedNum");
                successAccessedNum=(Integer) inMap.get("successAccessedNum");
                recordErrMsg=(String) inMap.get("recordErrMsg");
                events = (List<Event>) inMap.get("eventList");
                if(events == null){
                    LOG.error("因为发送基本数据单元集合(events)为空！所以不予发送;"+errMsg==null?"":errMsg);
                    errMsg="因为发送基本数据单元集合为空！所以不予发送;"+errMsg==null?"":errMsg;
                    finalMhd(request,response,out);
                }
                inMap.clear();
            } catch (HTTPBadRequestException ex) {
                if (StringUtils.isBlank(errMsg)) {
                    errMsg = "Received bad request from client！";
                }
                LOG.warn("Received bad request from client. ", ex);
                response.sendError(HttpServletResponse.SC_BAD_REQUEST,
                        "Bad request from client. "
                                + ex.getMessage());
                finalMhd(request, response, out);
                return;
            } catch (Exception ex) {
                if (StringUtils.isBlank(errMsg)) {
                    errMsg = "Deserializer threw unexpected exception.";
                }
                LOG.warn("Deserializer threw unexpected exception. ", ex);
                response.sendError(HttpServletResponse.SC_INTERNAL_SERVER_ERROR,
                        "Deserializer threw unexpected exception. "
                                + ex.getMessage());
                finalMhd(request, response, out);
                return;
            }
            sourceCounter.incrementAppendBatchReceivedCount();
            sourceCounter.addToEventReceivedCount(events.size());
            try {
                getChannelProcessor().processEventBatch(events);
            } catch (ChannelException ex) {
                //6表示推送给channel的时候发生错误
                status = 6;
                if (StringUtils.isBlank(errMsg)) {
                    errMsg = "Error appending event to channel. "
                            + "Channel might be full. Consider increasing the channel "
                            + "capacity or make sure the sinks perform faster.";
                }
                LOG.warn("Error appending event to channel. "
                        + "Channel might be full. Consider increasing the channel "
                        + "capacity or make sure the sinks perform faster.", ex);
                response.sendError(HttpServletResponse.SC_SERVICE_UNAVAILABLE,
                        "Error appending event to channel. Channel might be full."
                                + ex.getMessage());
                finalMhd(request, response, out);
                return;
            } catch (Exception ex) {
                //7表示在推送给channel的时候发生了未知异常
                status = 7;
                if (StringUtils.isBlank(errMsg)) {
                    errMsg = "Unexpected error appending event to channel.";
                }
                LOG.warn("Unexpected error appending event to channel. ", ex);
                response.sendError(HttpServletResponse.SC_INTERNAL_SERVER_ERROR,
                        "Unexpected error while appending event to channel. "
                                + ex.getMessage());
                finalMhd(request, response, out);
                return;
            }
            response.setCharacterEncoding(request.getCharacterEncoding()==null?"UTF-8":request.getCharacterEncoding());
            response.setStatus(HttpServletResponse.SC_OK);
            sourceCounter.incrementAppendBatchAcceptedCount();
            sourceCounter.addToEventAcceptedCount(events.size());
            StringBuffer sbf = new StringBuffer();
            sbf.append("{'ret_code':'").append(status).append("','err_msg':'")
                    .append(errMsg).append("','total_accessed_num':'")
                    .append(totalAccessedNum)
                    .append("','success_accessed_num':'")
                    .append(successAccessedNum)
                    .append("','recordErrMsg':'")
                    .append(recordErrMsg)
                    .append("'}");
            out.write(sbf.toString().getBytes("UTF-8"));
            out.close();
            response.flushBuffer();
        }

        @Override
        public void doGet(HttpServletRequest request, HttpServletResponse response)
                throws IOException {
            doPost(request, response);
        }
    }

    private static class HTTPSourceSocketConnector extends SslSocketConnector {
        private final List<String> excludedProtocols;

        HTTPSourceSocketConnector(List<String> excludedProtocols) {
            this.excludedProtocols = excludedProtocols;
        }

        @Override
        public ServerSocket newServerSocket(String host, int port, int backlog) throws IOException {
            SSLServerSocket socket = (SSLServerSocket) super.newServerSocket(host, port, backlog);
            String[] protocols = socket.getEnabledProtocols();
            List<String> newProtocols = new ArrayList<String>(protocols.length);
            for (String protocol : protocols) {
                if (!excludedProtocols.contains(protocol)) {
                    newProtocols.add(protocol);
                }
            }
            socket.setEnabledProtocols(newProtocols.toArray(new String[newProtocols.size()]));
            return socket;
        }
    }

    public void finalMhd(HttpServletRequest request, HttpServletResponse response, ServletOutputStream out) throws IOException {
        response.setCharacterEncoding(request.getCharacterEncoding());
        response.setStatus(HttpServletResponse.SC_INTERNAL_SERVER_ERROR);
        StringBuffer sbf = new StringBuffer();
        sbf.append("{'ret_code':'").append(status).append("','err_msg':'")
                .append(errMsg).append("','total_accessed_num':'")
                .append(totalAccessedNum)
                .append("','success_accessed_num':'")
                .append(successAccessedNum)
                .append("','recordErrMsg':'")
                .append(recordErrMsg)
                .append("'}");

        out.write(sbf.toString().getBytes("UTF-8"));
        response.flushBuffer();
        out.close();
    }
}
